package zredis

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/zouyang8210/zjson"
	"gitee.com/zouyang8210/zstrconv"
	"gitee.com/zouyang8210/ztools"
	"github.com/go-redis/redis/v8"
	"reflect"
	"time"
)

type ExpireEvent func(key string)

type ZRedis struct {
	cli           *redis.Client
	databaseName  string
	passwd        string
	addr          string
	db            int
	msgReadCount  int64
	msgBlock      int
	onExpireEvent ExpireEvent
}

const (
	PONG          = "PONG"
	OK            = "OK"
	Select        = "select"
	HmGetAll      = "HGETALL"
	ChannelExpire = "__keyevent@%d__:expired"
)

func (c *ZRedis) Connect(addr, pw string, db int) (e error) {
	c.addr = addr
	c.passwd = pw
	c.db = db
	c.msgReadCount = 10
	c.msgBlock = 0
	c.cli = redis.NewClient(&redis.Options{
		Addr:     c.addr,   // 地址
		Password: c.passwd, // 密码
		DB:       db,       // 默认数据库
		// DialTimeout:  10 * time.Second, // 连接超时
		// ReadTimeout:  3 * time.Second,  // 读超时
		// WriteTimeout: 3 * time.Second,  // 写超时
		// MinIdleConns: 10,               // 保留空闲连接
		// PoolSize:    10,               // 连接池大小
		// PoolTimeout: 30 * time.Second, // 连接池超时
	})
	if e = c.ping(); e != nil {
		return
	}

	return
}

// Select 选择数据库编号
func (c *ZRedis) Select(dbIndex int) (e error) {
	_, e = c.cli.Do(context.Background(), "select", dbIndex).Result()
	if e != nil {
		return
	}
	c.db = dbIndex
	return
}

// GetSelectIndex 获取当前所选择的数据库编号
func (c *ZRedis) GetSelectIndex() int {
	return c.db
}

// SetExpireEvent 设置键过期事件
func (c *ZRedis) SetExpireEvent(event ExpireEvent) {
	if c.onExpireEvent == nil {
		c.onExpireEvent = event
	}
}

func (c *ZRedis) Exist(key string, ) (b bool, e error) {
	var res int64
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	res, e = c.cli.Exists(context.Background(), []string{key}...).Result()
	if e != nil {
		return
	}
	if res > 0 {
		return true, nil
	}
	return
}

func (c *ZRedis) Set(key string, v interface{}, timeout int) (e error) {
	var res string
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	res, e = c.cli.Set(context.Background(), key, v, time.Duration(timeout)*time.Second).Result()
	if e != nil {
		return
	}
	if res != OK {
		e = errors.New(res)
	}
	return
}

func (c *ZRedis) GetUint(key string) (v uint64, e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	var res string
	if res, e = c.cli.Get(context.Background(), key).Result(); e != nil {
		return
	}
	e = zstrconv.StrToUint(res, &v)
	return
}

func (c *ZRedis) GetInt(key string) (v int64, e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	var res string
	if res, e = c.cli.Get(context.Background(), key).Result(); e != nil {
		return
	}
	e = zstrconv.StrToInt(res, &v)
	return
}

func (c *ZRedis) GetFloat(key string) (v int64, e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	var res string
	if res, e = c.cli.Get(context.Background(), key).Result(); e != nil {
		return
	}
	e = zstrconv.StrToFloat(res, &v)
	return
}

func (c *ZRedis) GetString(key string) (v string, e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	v, e = c.cli.Get(context.Background(), key).Result()
	return
}

func (c *ZRedis) Del(key string) (e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	_, e = c.cli.Del(context.Background(), key).Result()
	if e != nil {
		return
	}
	return

}

// ===========================================消息列队===================================================

// CreateConsumerGroup 创建消费者
func (c *ZRedis) CreateConsumerGroup(streamKey, consumerGroup string) (e error) {
	var group []redis.XInfoGroup
	if group, e = c.cli.XInfoGroups(context.Background(), streamKey).Result(); e == nil {
		if len(group) > 0 {
			return nil
		}
		if e = c.cli.XGroupCreate(context.Background(), streamKey, consumerGroup, "0-0").Err(); e != nil {
			return
		}
		return
	}
	if e = c.cli.XGroupCreate(context.Background(), streamKey, consumerGroup, "0-0").Err(); e != nil {
		return
	}
	return
}

// AddMessageList 消息列列队添加消息
func (c *ZRedis) AddMessageList(streamKey string, msg interface{}) (msgID string, e error) {
	var mapMsg map[string]interface{}
	if e = zjson.ObjToObjNoNumber(&mapMsg, msg); e != nil {
		return
	}
	msgID, e = c.cli.XAdd(context.Background(), &redis.XAddArgs{
		Stream: streamKey,
		Values: mapMsg,
	}).Result()
	if e != nil {
		return
	}
	return
}

type Message struct {
	Id      string
	Content string
}

// ReadMessageList 读取消息列队
func (c *ZRedis) ReadMessageList(streamKey, consumerGroup, consumerName string) (message []Message, e error) {
	var streams []redis.XStream
	streams, e = c.cli.XReadGroup(context.Background(), &redis.XReadGroupArgs{
		Group:    consumerGroup,
		Consumer: consumerName,
		Streams:  []string{streamKey, "0-0"}, // ">" 代表只读取新消息 "0-0" 从流的第一条消息开始
		Count:    c.msgReadCount,
		Block:    time.Duration(c.msgBlock) * time.Second,
	}).Result()
	if e != nil {
		return
	}
	for _, stream := range streams {
		for _, msg := range stream.Messages {
			m := Message{}
			m.Id = msg.ID
			m.Content, _ = zjson.ToJson(msg.Values)
			message = append(message, m)
		}
	}
	return
}

// AckMessage 确认消息已处理
func (c *ZRedis) AckMessage(streamKey, consumerGroup string, msgID string, del bool) (e error) {
	_, e = c.cli.XAck(context.Background(), streamKey, consumerGroup, msgID).Result()
	if e == nil && del {
		_, e = c.cli.XDel(context.Background(), streamKey, msgID).Result()
	}
	return
}

//=========================================================END=======================================================

// HMSet 保存结构，obj是用一结构
func (c *ZRedis) HMSet(key string, obj interface{}) (e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()

	mType := reflect.TypeOf(obj)
	mValue := reflect.ValueOf(obj)
	numField := mType.NumField()

	var m = make(map[string]interface{})

	for i := 0; i < numField; i++ {
		m[mType.Field(i).Name] = mValue.Field(i).Interface()
	}

	_, e = c.cli.HMSet(context.Background(), key, m).Result()

	return
}

// HMGet 获取结构信息，v是一个结构，需已保存的结构一致
func (c *ZRedis) HMGet(key string, v interface{}) (e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()

	res, e := c.cli.Do(context.Background(), HmGetAll, key).Result()
	if e != nil {
		return
	}
	fieldsValue := res.([]interface{})
	fieldsNum := len(fieldsValue)
	if fieldsNum <= 0 {
		return errors.New("key not found")
	}
	mValue := reflect.ValueOf(v).Elem()

	for i := 0; i < fieldsNum; i += 2 {
		field := fieldsValue[i].(string)
		if !mValue.FieldByName(field).CanSet() {
			e = errors.New(field + " can't set value")
			return
		}
		value := fieldsValue[i+1].(string)
		switch mValue.FieldByName(field).Type().String() {
		case "string":
			mValue.FieldByName(field).SetString(value)
		case "int8", "int", "int32", "int16", "int64":
			var iTmp int64
			if e = zstrconv.StrToInt(value, &iTmp); e != nil {
				return
			}
			mValue.FieldByName(field).SetInt(iTmp)
		case "uint8", "uint", "uint32", "uint16", "uint64":
			var iTmp uint64
			if e = zstrconv.StrToUint(value, &iTmp); e != nil {
				return
			}
			mValue.FieldByName(field).SetUint(iTmp)
		case "float32", "float64":
			var iTmp float64
			if e = zstrconv.StrToFloat(value, &iTmp); e != nil {
				return
			}
			mValue.FieldByName(field).SetFloat(iTmp)
		case "bool":
			var iTmp uint8
			if e = zstrconv.StrToUint(value, &iTmp); e != nil {
				return
			}
			x := ztools.If(iTmp > 0, true, false).(bool)
			mValue.FieldByName(field).SetBool(x)
		case "time.Time":
			var iTmp time.Time
			iTmp = zstrconv.ToLocalTime(value)
			if !iTmp.IsZero() {
				mValue.FieldByName(field).Set(reflect.ValueOf(iTmp))
			}
		}
	}
	return
}

func (c *ZRedis) HMGetFields(key string, v ...string) (m map[string]string, e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	fieldsValue, e := c.cli.HMGet(context.Background(), key, v...).Result()
	if e != nil {
		return
	}

	fieldsNum := len(fieldsValue)
	if fieldsNum <= 0 {
		return nil, errors.New("key not found")
	}

	m = make(map[string]string)
	for i := 0; i < fieldsNum; i++ {
		if fieldsValue[i] == nil {
			continue
		}
		m[v[i]] = fieldsValue[i].(string)
	}
	return
}

// Expire 设置键超时(秒)
func (c *ZRedis) Expire(key string, timeout int) (e error) {
	if e = c.Lock(); e != nil {
		return
	}
	defer c.unlock()
	b, e := c.cli.Expire(context.Background(), key, time.Duration(timeout)*time.Second).Result()
	if e != nil {
		return
	}
	if !b {
		e = errors.New("set expire failed")
	}
	return
}

// SubscribeExpired 订阅键超时,需先调用SetExpireEvent设置事件
func (c *ZRedis) SubscribeExpired() (e error) {
	if c.onExpireEvent == nil {
		return errors.New("no set onExpireEvent")
	}
	go func() {
		ctx := context.Background()
		pub := c.cli.Subscribe(ctx, fmt.Sprintf(ChannelExpire, c.db))
		for {
			ch := pub.Channel()
			for msg := range ch {
				//fmt.Printf("收到消息: %s, 通道: %s, 消息内容: %s\n", msg.Payload, msg.Channel, msg.Pattern)
				if c.onExpireEvent != nil {
					c.onExpireEvent(msg.Payload)
				}
			}
		}
	}()
	return
}

func (c *ZRedis) ping() (e error) {
	var res string
	res, e = c.cli.Ping(context.Background()).Result()
	if e != nil {
		return
	}
	if res != PONG {
		e = errors.New(res)
	}
	return
}

var chLock = make(chan int, 1)

func (c *ZRedis) Lock() (e error) {
	chLock <- 1
	//if e = c.ping(); e != nil {
	//	return
	//}
	return
}

func (c *ZRedis) unlock() {
	<-chLock
}
